package com.shujia.spark.core

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel

object Demo16Cache {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setAppName("Demo16Cache").setMaster("local")

    val sc = new SparkContext(conf)


    //设置checkpoint路径
    sc.setCheckpointDir("spark/data/checkpoint")


    /**
      * 1、统计班级人数
      * 2、统计性别人数
      */
    val studentsRDD = sc.textFile("spark/data/students.txt")

    val stuRDD: RDD[Student] = studentsRDD.map(line => {
      val split = line.split(",")
      println(split(0))
      Student(split(0), split(1), split(2).toInt, split(3), split(4))
    })

    //对多次使用RDD进行缓存
    //cache  将数据缓存到内存中
       stuRDD.cache()

    /**
      * StorageLevel 持久化级别
      * 持久化级别选择
      * 1、内存充足-->  MEMORY_ONLY （直接放内存）  (如果内存不足会出现内存溢出)
      * 2、内存一般-->  MEMORY_ONLY_SER（将数据压缩后放内存）
      * 3、数据量很大 -->  MEMORY_AND_DISK_SER (压缩之后放内存，内存放不下放磁盘)  为什么不使用MEMORY_AND_DISK  （不管压不压缩放内存都比放磁盘快）
      *
      * 解压和压缩需要消耗cpu的时间
      *
      * 如果缓存到磁盘（磁盘：task执行所在节点的磁盘， 不需要指定缓存的路径，任务执行完成会自动删除）
      * 内存：task执行所在节点的内存
      */

    //持久化级别
    //stuRDD.persist(StorageLevel.MEMORY_ONLY)


    /**
      * checkpoint  可以将当前rdd的数据保存到hdfs    ，数据可以永久保留
      * 需要在SparkContext中设计Checkpoint路径（本地路径或者hdfs路径）
      *
      * 当第一个job执行完成之后，会从最后一个RDD向前回溯，如果发现哪一个RDD使用了checkpoint ，会被打上一个标记。
      * 另启动一个job任务重新计算这个rdd的数据，并将数据保存到hdfs 。所以stuRDD在这里会被计算两次
      * 所以可以对做了checkpoint的RDD先做cache,就不会重复计算了
      * checkpoint会切断RDD之间的依赖关系， 后续job的执行直接去读取hdfs中的数据进行计算
      *
      */

    //再checkpoint可以先进行cache
    stuRDD.cache()

    //将stuRDD的数据保存到hdfs
    stuRDD.checkpoint()


    //统计班级人数
    val clazzKVRDD: RDD[(String, Int)] = stuRDD.map(stu => (stu.clazz, 1))
    val clazzNumRDD: RDD[(String, Int)] = clazzKVRDD.reduceByKey((x, y) => x + y)
    clazzNumRDD.map(x => x._1 + "\t" + x._2).saveAsTextFile("spark/data/clazz_num")


    //统计性别人数
    val genderKVRDD: RDD[(String, Int)] = stuRDD.map(stu => (stu.gender, 1))
    val genderNumRDD: RDD[(String, Int)] = genderKVRDD.reduceByKey((x, y) => x + y)
    genderNumRDD.map(x => x._1 + "\t" + x._2).saveAsTextFile("spark/data/gender_num")


    val genderKVRDD1: RDD[(String, Int)] = stuRDD.map(stu => (stu.gender, 1))
    val genderNumRDD1: RDD[(String, Int)] = genderKVRDD1.reduceByKey((x, y) => x + y)
    genderNumRDD1.map(x => x._1 + "\t" + x._2).saveAsTextFile("spark/data/gender_num1")


  }

  case class Student(id: String, name: String, age: Int, gender: String, clazz: String)

}
